package com.vagavaga.tcpt;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedByInterruptException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

import org.apache.log4j.Logger;

import com.vagavaga.tcpt.TransportServer.Connection;


class SelectorThread extends Thread {
    private static final Logger log = Logger.getLogger(SelectorThread.class);

    private final TransportServer server;
    private final Selector selector;
    private ByteBuffer readingBuffer = ByteBuffer.allocateDirect(1024 * 10);

    public SelectorThread(TransportServer server, Selector selector) {
        if (server == null) throw new NullPointerException("server");
        if (selector == null) throw new NullPointerException("selector");
        this.server = server;
        this.selector = selector;
    }

    @Override
    public void run() {
        log.info("Starting Selector Thread");
        try {

            while (!isInterrupted()) {
                int keys = 0;
                try {
                    keys = selector.select(100);
                } catch (NullPointerException npe) {
                    // to catch selector bug;
                }

                if (keys == 0) {
                    continue;
//                    if (selector.keys().isEmpty() && isInterrupted()) {
//                        break;
//                    }
                }

                // do operations on selected set.
                for (Iterator<SelectionKey> it = selector.selectedKeys().iterator(); it.hasNext();) {
                    SelectionKey key = it.next();

                    try {
                        handleKey(key);
                    } catch (CancelledKeyException cke) {

                        log.debug("Catched access to closed object [" + key.attachment() + "]", cke);
                        Object att = key.attachment();
                        if (att instanceof Connection) server.fireDisconnected((Connection) att);

                    } catch (IOException ex) {
                        Object att = key.attachment();
                    
                        if (((ex instanceof ClosedChannelException) || (ex instanceof ClosedByInterruptException))
                                && att instanceof Connection) {
                            log.debug("Catched channel close event [" + att + "]", ex);
                        } else {
                            log.info("IOError occured!", ex);
                            Connection conn = null;
                            if (att instanceof Connection) {
                                conn = (Connection) att;
                            }
                            server.fireError(conn, ex);
                        }
                        server.fireDisconnected((Connection) att);
                        if (key.isValid()) {
                            key.channel().close();
                        }
                    } catch (Throwable th) {
                        log.error("Severe unhandled error occured", th);
                    } finally {
                        it.remove();
                    }
                }
            }

//            selector.selectNow();
//            for (Iterator it = selector.keys().iterator(); it.hasNext();) {
//                SelectionKey k = (SelectionKey) it.next();
//                Object att = k.attachment();
//                if (att instanceof IConnection) {
//                    server.fireDisconnected((IConnection) att);
//                }
//            }
        } catch (IOException e) {
            log.error("Selector thread cauth unhandled exception", e);
        } catch (ThreadDeath death) {
            log.info("Thread Death occured");
            throw death;
        } finally {
            log.debug("Cleanup");
            try {
                if (selector.isOpen()) selector.close();
            } catch (IOException ioe) {
            }
        }
        log.debug("Selector Thread about to die");
    }

    private void handleKey(SelectionKey key) throws IOException {

        if (key.isReadable()) read(key);
        if (key.isWritable()) write(key);
        if (key.isAcceptable()) accept(key);
        if (key.isConnectable()) connect(key);

    }

    private void connect(SelectionKey key) throws IOException {
        SocketChannel ch = (SocketChannel) key.channel();
        Connection connection = (Connection) key.attachment();
        if (ch.finishConnect()) {
            key.interestOps(key.interestOps() & (~SelectionKey.OP_CONNECT));
            server.fireConnected(connection);
            key.interestOps(key.interestOps() | SelectionKey.OP_READ);
        }
    }

    private void read(SelectionKey key) throws IOException {
        SocketChannel ch = (SocketChannel) key.channel();
        IConnection conn = (IConnection) key.attachment();
        log.debug("Reading data from socket [" + conn + "]");
        readingBuffer.clear();
        int read = 0;
        while ((read = ch.read(readingBuffer)) > 0) {
            readingBuffer.flip();
            server.fireDataReceived(conn, readingBuffer.asReadOnlyBuffer());
            readingBuffer.clear();
        }
        if (read == -1) {
            log.debug("Channel will be closed since EOF was read");
            key.channel().close();
        }
    }

    private void write(SelectionKey key) throws IOException {
        SocketChannel ch = (SocketChannel) key.channel();
        Connection conn = (Connection) key.attachment();
        log.debug("Writting data to socket [" + conn + "]");

        synchronized (conn.sendingQueue) {
            if (conn.sendingQueue.isEmpty()) return;
            ByteBuffer writtingBuffer = conn.sendingQueue.getFirst();

            ch.write(writtingBuffer);

            if (!writtingBuffer.hasRemaining()) {
                conn.sendingQueue.removeFirst();
                if (conn.sendingQueue.isEmpty()) {
                    log.debug("No more data to write [" + conn + "]");
                    key.interestOps(key.interestOps() & (~SelectionKey.OP_WRITE));
                }
                conn.sendingQueue.notifyAll();
            }
        }
    }

    private void accept(SelectionKey key) throws IOException {
        log.debug("Accepting new connection");
        ServerSocketChannel ch = (ServerSocketChannel) key.channel();
        SocketChannel client = ch.accept();
        client.configureBlocking(false);
        Connection conn = server.createConnection();
        log.debug("Connected [" + conn + "]");
        server.fireConnected(conn);
        conn.key = client.register(selector, SelectionKey.OP_READ, conn);
    }
}